package com.niit.spark.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

/**
 * Date:2025/4/14
 * Author：Ys
 * Description:
 */
object RDD_Transform01 extends App {

  //1.准备环境
  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
  //2.设置sc
  val sc = new SparkContext(sparkConf)


  sc.setLogLevel("ERROR")

  /*
  map:将RDD中的每个元素应用一个函数，并将函数的结果作为新的RDD
      将处理的数据逐条进行映射转换。这里的转换可以是类型的转换，也可以是值得转换
    需求：将RDD中的 1 2 3 4 转换 2  4 6 8
   */
  val rdd1 = sc.makeRDD( List(1,2,3,4) )
  val mapRdd1: RDD[Int] = rdd1.map(_ * 2)
  mapRdd1.collect().foreach(println)
  println(mapRdd1.collect().mkString(" "))

  /*
  mapPartitions:将待处理的数据以分区为单位发送道计算节点进行处理。
   */
  val rdd2 = sc.makeRDD( List(1,2,3,4) ,2)//【1，2】  2   【3 ，4】 4
  //需求：每个分区内的最大值
  val mapRdd2: RDD[Int] = rdd2.mapPartitions(iter => {
    List(iter.max).iterator
  })
  mapRdd2.collect().foreach(println)

  /*
  flatMap:将RDD中进行扁平映射，将RDD中的每个元素应用一个函数，并将函数的结果作为新的RDD
   */
  val rdd3 = sc.makeRDD(   List(   List(1,2),List(3,4),List(5,6)   ) )
  val mapRdd3: RDD[Int] = rdd3.flatMap(list => list)
  mapRdd3.collect().foreach(println)
  //拆分 RDD中的单词
  val rdd4 = sc.makeRDD(   List(   "hello world",  "hello scala",  "hello spark"   ) )
  //==>List(hello,world,hello,scala,hello,spark)
  val maoRdd4: RDD[String] = rdd4.flatMap(_.split(" "))
  maoRdd4.collect().foreach(println)

  /*
  glom:将同一个分区的数据直接转换为相同类型的内存数组进行处理，分区不变
   */
  val rdd5 = sc.makeRDD( List(1,2,3,4) ,2)//【1，2】     【3 ，4】
  val glomRdd1: RDD[Array[Int]] = rdd5.glom()// [ [1,2],[3,4] ]
  glomRdd1.collect().foreach(  data=> println(data.mkString(",")) )

  //需求：获得两个分区的最大值的和  2 + 4
  val glomRdd2: RDD[Array[Int]] = rdd5.glom()
  val maxRdd: RDD[Int] = glomRdd2.map(arr => {
    arr.max
  })//[2,4]
  println( maxRdd.reduce(_ + _) )

  /*
  groupBy:将数据按照指定的规则进行分组，分区默认不变。但是数据会被打乱重新组合。我们将这样的操作称之为shuffle.
          极限的情况下，数据可能被分仔同一个分区中， 一个组的数据在一个分区中，但是并不可以说一个分区中只有一个组
     */
  val rdd7 = sc.makeRDD( List(1,2,3,4) ,2)//奇数一组，偶数一组
  val groupRdd1: RDD[(Int, Iterable[Int])] = rdd7.groupBy(_ % 2)
  groupRdd1.collect().foreach(println)

  //需求：按照单词的首字母进行分组
  val rdd8 = sc.makeRDD(   List( "Hello","Spark","HBase","Scala","Kafka" ) )
  // Hello -- > H  --> "Hello".chartAt(0)
  val groupRdd3: RDD[(Char, Iterable[String])] = rdd8.groupBy(_.charAt(0))
  groupRdd3.collect().foreach(println)

  /*
  filter 将数据根据指定的规则进行筛选过滤，符合规则的留下，不符合规则的丢弃
       数据进行筛选过滤后，分区保持不变，但是分区内的数据可能不均衡，在生产环境下，可能出现数据不均衡的问题
   需求：获得列表中的奇数
   */
  val rdd9: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8, 9))
  val filterRDD: RDD[Int] = rdd9.filter(_ % 2 != 0)
  filterRDD.collect().foreach(println)

  /*
   distinct:对RDD中的数据进行去重，并返回一个新的RDD
   */
  val rdd10 = sc.makeRDD( List(1,2,3,4,5,6,7,8,9,1,2,3,4,5,6,7,8,9) )
  val distinctRDD: RDD[Int] = rdd10.distinct(2)// 2是分区数，去重后的数据重新分区
  println("分区数：",distinctRDD.getNumPartitions)
  distinctRDD.collect().foreach(println)

  /*
  sortBy：默认是一个升序，如果想改成降序，第二个参数设置为false
   */
  val rdd11: RDD[Int] = sc.makeRDD(List(3, 1, 2, 4, 5))
  val sortRdd1: RDD[Int] = rdd11.sortBy(num => num, false)
  sortRdd1.collect().foreach(println)

  //需求：按照key进行排序，升序
  val rdd12: RDD[(String, String)] = sc.makeRDD(List(  ("1", "a"), ("3", "c"), ("2", "b"), ("4", "d")))
  val sortRdd2: RDD[(String, String)] = rdd12.sortBy(_._1.toInt)
  sortRdd2.collect().foreach(println)

  /*
  coalesce :根据数据缩减分区，用于大数据集过滤后，提高小数据集的执行效率
   当Spark程序中，存在过多的小任务的时候，可以通过coalesce操作，将数据缩减分区，减少小任务的数量，减少任务调度成本
   */
  val rdd13: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6), 3)
  //【1,2】 【3，4】 【5，6】== > 【1，2，3】 【4，5，6】
  //减小分区的时候会发生数据倾销，如何避免数据倾斜？
  //将第二个参数设置为true，会进行shuffle操作，将数据进行重新分区
  val rdd14: RDD[Int] = rdd13.coalesce(2, true)
 //rdd14.saveAsTextFile("output5")

  /*
  repation:该操作内部其实执行的是coalesce，参数shuffle为true。
  无论是将分区数变少变多，repation都可以完成
   */
  private val reRdd: RDD[Int] = rdd13.repartition(6)
  //reRdd.saveAsTextFile("output6")

  /*
   RDD的交集 并集 差集  拉链
   */
  val rdd15 = sc.makeRDD(List(1,2,3,4))
  val rdd16 = sc.makeRDD(List(3,4,5,6))

  //并集
  val unRdd: RDD[Int] = rdd15.union(rdd16)
  println(unRdd.collect().mkString(","))

  //差集
  val subRdd: RDD[Int] = rdd15.subtract(rdd16)
  println(subRdd.collect().mkString(","))

  //交集
  val interRdd: RDD[Int] = rdd15.intersection(rdd16)
  println(interRdd.collect().mkString(","))

  //拉链
  val zipRdd: RDD[(Int, Int)] = rdd15.zip(rdd16)
  println(zipRdd.collect().mkString(","))

  /*
  partitionBy:根据指定的规则进行分区，Spark的默认分区器是HashPartitioner
   */
  val rdd17 = sc.makeRDD( List(1,2,3,4) , 2 )
  // ==> (1,1) (2,1) (3,1) (4,1)
  // 隐式转换
  val mapRdd4: RDD[(Int, Int)] = rdd17.map((_, 1))
  val parRdd: RDD[(Int, Int)] = mapRdd4.partitionBy(new HashPartitioner(3))
  //parRdd.saveAsTextFile("output7")

  /*
    reduceByKey: 可以将数据按照相同的Key对Value进行聚合,所以Key是不参与运算
   */
  val rdd18 = sc.makeRDD( List(  ("a",1),("a",2),("b",3),("b",4),("b",5),("a",6)  ) )
   val redRdd: RDD[(String, Int)] = rdd18.reduceByKey((x: Int, y: Int) => {
    x + y
  })//(a,9) b(b,12)
  val redRdd2: RDD[(String, Int)] = rdd18.reduceByKey(_ + _)
  redRdd2.collect().foreach(println)


  /*
   groupByKey:可以将数据按照相同的Key进行分组,Key不参与计算
 */
  //根据Key对value进行分组

   val groupRdd2: RDD[(String, Iterable[Int])] = rdd18.groupByKey()
  //==>[ (a,[1,2,6] ) ,(b,[3,4,5] ) ]
  groupRdd2.collect().foreach(println)


  /*
   aggregateByKey:将数据按照相同的Key进行聚合， 聚合==>将数据根据不同的规则进行分区内和分区间的计算
   */
  // val rdd18 = sc.makeRDD( List(  ("a",1),("a",2),("b",3),("b",4),("b",5),("a",6)  ) )
  // 找每个分区内的最大值后进行求和
  //参数1：表示初始值
  //参数2： 需要传递两个参数，第一个参数是分区内的计算规则，第二个参数是分区间的计算规则
  val rdd19 = sc.makeRDD( List(  ("a",1),("a",2),("b",3),  ("b",4),("b",5),("a",6)  ) ,2 )
  val agRdd: RDD[(String, Int)] = rdd19.aggregateByKey(0)(
    (x, y) => math.max(x, y),//找到每个Key中的最大值
    (x, y) => x + y // x 6 + y 5
  )
  agRdd.collect().foreach(println)
  /*
    ByKey==>  【("a",1),("a",2),("b",3)】  【 ("b",4),("b",5),("a",6) 】
              【 a: 1 2   b : 3】           【 b: 4 5   a : 6  】
   第一步：分区内找道最大值  根据不同的Key找到最值
          【 a: 2   b : 3】           【 b: 5   a : 6  】
   第二步： 分区间进行求和 根据相同的Key ，对值进行求和
         a 2 + a 6 = a  8
         b 3 + b 5 = b  8
   */

  //rrd19 根据Key进行分区内求和后进行分区间求和
  val agRdd2: RDD[(String, Int)] = rdd19.aggregateByKey(0)(
    (x, y) => x+y, //找到每个Key中的最大值
    (x, y) => x + y // x 6 + y 5
  )
  //简写
  val agRdd3: RDD[(String, Int)] = rdd19.aggregateByKey(0)(_ + _, _ + _)
  //再简写： 如果分区内和分区间的计算规则是相同，可以使用foldByKey
  val agRdd4: RDD[(String, Int)] = rdd19.foldByKey(0)(_ + _)

  //rrd19 根据Key 计算平均值  要知道每个Key出现的次数 以及每个Key的Value进行相加
  //val rdd19 = sc.makeRDD( List(  ("a",1),("a",2),("b",3),  ("b",4),("b",5),("a",6)  ) ,2 )
  // (0,0) 第一个 t._1 0 表示 Value初始值  第二个  t._2 0 表示key出现次数初始值
  //
  private val agRdd5: RDD[(String, (Int, Int))] = rdd19.aggregateByKey((0, 0))(
    (t, v) => { //t 是一个元组，(0,0)  v 就是key对应的值
      (t._1 + v, t._2 + 1) //==> t._1=0 Value   t._1 + 1 + 2  t._2 = 0 Key  t._2 + + 1  第一个分区内计算结果 a (3,2)  b (3,1)
      // 第二个分区内计算结果  a (6,1) b(9 , 2)
    },
    (t1, t2) => { // t1 第一个分区内的结果  t2 第二个分区内结果
      // ByKey
      (t1._1 + t2._1, t1._2 + t2._2) // a: ( 3 + 6 , 2 + 1) => (9,3)   b:(3 + 9, 1 + 2) => (12,3)
    }
  )

  val resRdd:RDD[(String,Int)] = agRdd5.mapValues{
    case (sum, count)=>{
      sum/count  // a 9/ 3         b 12 / 3
    }
  } //==>mapByKey
                                              // t 就是 agRdd5   (Int, Int) ==>  (9,3)   (12,3)
   val resRdd2: RDD[(String, Int)] = agRdd5.mapValues(t => t._1 / t._2)

  resRdd.collect().foreach(println)

  /*
  sortByKey:根据Key进行排序,默认是升序，参数为false为降序
   */
  val rdd20 = sc.makeRDD( List(  ("a",1) ,("c",3), ("b",2) ) )
  //==> (a,1) (b,2) (c,3)
  val sortkeyRdd: RDD[(String, Int)] = rdd20.sortByKey(false)
  sortkeyRdd.collect().foreach(println)

  /*
    join :根据Key进行Join，会进行交集的过滤，返回的RDD的Key为交集的Key，Value为元组，元组中第一个为第一个RDD的Value，第二个为第二个RDD的Value
   */

  val rdd21 = sc.makeRDD( List(  ("a",1),("a",2),("c",3),("b",8) ) ) //rdd21 主键  a    c     MySQL 内连接  inner join  user.sid=student.id
  val rdd22 = sc.makeRDD( List(  ("a",5),("c",6),("a",4),("g",7) ) ) // rdd22 外键 a    c
  private val joinRdd: RDD[(String, (Int, Int))] = rdd21.join(rdd22)
  joinRdd.collect().foreach(println)

  val joinRdd2: RDD[(String, (Int, Option[Int]))] = rdd21.leftOuterJoin(rdd22)
  joinRdd2.collect().foreach(println)

  val joinRdd3: RDD[(String, (Option[Int], Int))] = rdd21.rightOuterJoin(rdd22)
  joinRdd3.collect().foreach(println)

  /*
   cogroup:分组后拼接 (connect+group)
   */
  val rdd23 = sc.makeRDD( List(  ("a",1),("b",2)/*,("c",3)*/ ) )
  val rdd24 = sc.makeRDD( List(  ("a",4),("b",5),("c",6),("c",7) ) )
  /*
   a :1 4
   b:2,5
   c:3,6,7
   */
  val cogroupRdd: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd23.cogroup(rdd24)
  cogroupRdd.collect().foreach(println)


  //3.关闭环境
  sc.stop()
}
